package com.atbeijing.D03;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import com.atbeijing.util.DateUtil;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 自定义窗口聚合函数,处理滚动时间窗口数据
 * 先增量函数聚合:不需要收集窗口中的所有元素，只需要维护一个累加器，节省内存
 * 再全量函数聚合:为结果加上窗口信息
 *
 * 求5s内每个温感器的平均温度,附加窗口信息
 */
public class Example9 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.id)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new Agg1(),new windowResult1())
                .print();

        env.execute();
    }

    /**
     * 每条流聚合到累加器
     */
    public static class Agg1 implements AggregateFunction<SensorReading, Tuple3<String,Double,Long>, Tuple2<String,Double>>{
        @Override
        public Tuple3<String, Double, Long> createAccumulator() {
            return Tuple3.of("",0.0, 0L);
        }

        @Override
        public Tuple3<String, Double, Long> add(SensorReading value, Tuple3<String, Double, Long> accumulator) {
            accumulator.f0=value.id;
            accumulator.f1=value.temperature;
            accumulator.f2=1L;
            return accumulator;
        }

        @Override
        public Tuple2<String, Double> getResult(Tuple3<String, Double, Long> accumulator) {
            return Tuple2.of(accumulator.f0,accumulator.f1/accumulator.f2);
        }

        @Override
        public Tuple3<String, Double, Long> merge(Tuple3<String, Double, Long> a, Tuple3<String, Double, Long> b) {
            return null;
        }
    }

    /**
     * 接收每条流累加器处理,每条流的窗口关闭时只有一个累加器下发,所以iterable中只有一个数据,算好的平均值
     */
    public static class windowResult1 extends ProcessWindowFunction<Tuple2<String, Double>, Tuple4<String,Double,String,String>,String, TimeWindow>{
        //
        @Override
        public void process(String s, Context context, Iterable<Tuple2<String, Double>> iterable, Collector<Tuple4<String, Double, String, String>> collector) throws Exception {
            String s1 = DateUtil.milliTimestampToDate(context.window().getStart());
            String s2 = DateUtil.milliTimestampToDate(context.window().getEnd());
            collector.collect(Tuple4.of(s,iterable.iterator().next().f1,s1,s2));
        }
    }
}
